package cn.dapeng.rpc.client;

import cn.dapeng.rpc.client.pool.ClientFactoryV1;
import cn.dapeng.rpc.common.MyContent;
import cn.dapeng.rpc.common.MyHeader;
import cn.dapeng.rpc.common.RpcResponse;
import cn.dapeng.rpc.common.cb.ResponseCallBackV2;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

public class ClientV1 {

    public static RpcResponse call(String s) {
        ProxyCall proxyCall = proxyGet(ProxyCall.class);
        return proxyCall.call(s);
    }

    public static <T> T proxyGet(Class<T> clazz) {

        Class<?>[] methodInfo = {clazz};
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), methodInfo, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

                // 协议：  header+msgBody
                MyContent content = new MyContent(clazz.getName(), method.getName(), method.getParameterTypes(), args);

                ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                ObjectOutputStream oos = new ObjectOutputStream(outputStream);
                oos.writeObject(content);
                byte[] msgBody = outputStream.toByteArray();

                MyHeader header = createHeader(msgBody);

                outputStream.reset();
                oos = new ObjectOutputStream(outputStream);
                oos.writeObject(header);
                byte[] msgHeader = outputStream.toByteArray();

                System.out.println("msgHeader length : " + msgHeader.length);

                // 连接池
//                CountDownLatch countDownLatch = new CountDownLatch(1);
//                ResponseCallBackV1.addCallback(header.getRequestId(), () -> {
//                    countDownLatch.countDown();
//                });
                CompletableFuture future = new CompletableFuture();
                ResponseCallBackV2.addCallback(header.getRequestId(), future);
                NioSocketChannel clientChannel = ClientFactoryV1.getInstance().getClient(new InetSocketAddress("localhost", 9090));

                ByteBuf sendBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgBody.length + msgHeader.length);
                sendBuf.writeBytes(msgHeader);
                sendBuf.writeBytes(msgBody);
                clientChannel.writeAndFlush(sendBuf).sync();

                //countDownLatch.await();

                return future.get();
            }
        });
    }

    public static MyHeader createHeader(byte[] body) {
        int f = 0x14141414;
        int size = body.length;
        long requestId = Math.abs(UUID.randomUUID().getLeastSignificantBits());
        MyHeader header = new MyHeader();
        header.setFlag(f);
        header.setRequestId(requestId);
        header.setBodyLen(size);
        return header;
    }

}
